#include "ThreadPool.h"

using namespace cvs;

ThreadPool::ThreadPool(const std::string& name)
	: name_(name)
	, running_(false)
{

}

ThreadPool::~ThreadPool() {
	if (running_) {
		stop();
	}
}

void ThreadPool::start(int numThreads) {
	assert(numThreads>0);
	// threads have been started
	if (!threads_.empty()) {
		return;
	}

	running_ = true;
	threads_.reserve(numThreads);

	for (int index=0; index<numThreads; ++index) {
		char id[32];
		sprintf(id, "%d", index);

		threads_.push_back(new cvs::Thread(boost::bind(&ThreadPool::runInThread, this), name_+id));
		threads_[index]->start();
	}
}

void ThreadPool::stop() {
	//running_ = false;
	cond_.notify_all();

	printf("still not finished task num: %d\n", queue_.size());

	// block wait all the threads stopped
	for_each(threads_.begin(), threads_.end(), boost::bind(&cvs::Thread::join, _1));
}

void ThreadPool::run(const Task& task) {
	if (threads_.empty()) {
		task();
	} else {
		boost::mutex::scoped_lock lock(mutex_);
		queue_.push_back(task);

		cond_.notify_one();
	}
}

ThreadPool::Task ThreadPool::take(void) {
	boost::mutex::scoped_lock lock(mutex_);

	while(queue_.empty() && running_) {
		cond_.wait(mutex_);
	}

	ThreadPool::Task task;
	if (!queue_.empty()) {
		task = queue_.front();
		queue_.pop_front();
	} 

	return task;
}


void ThreadPool::runInThread() {
	try
	{
		while (running_)
		{
			Task task(take());
			if (task)
			{
				task();
			}
			if (queue_.empty()) return; 
		}
	}
	//catch (const Exception& ex)
	//{
	//	fprintf(stderr, "exception caught in ThreadPool %s\n", name_.c_str());
	//	fprintf(stderr, "reason: %s\n", ex.what());
	//	fprintf(stderr, "stack trace: %s\n", ex.stackTrace());
	//	abort();
	//}
	catch (const std::exception& ex)
	{
		fprintf(stderr, "exception caught in ThreadPool %s\n", name_.c_str());
		fprintf(stderr, "reason: %s\n", ex.what());
		abort();
	}
	catch (...)
	{
		fprintf(stderr, "unknown exception caught in ThreadPool %s\n", name_.c_str());
		abort();
	}
}